use std::{any::Any, sync::Arc};

use futures::Future;
use async_trait::async_trait;


#[derive(Debug, thiserror::Error, PartialEq)]
pub enum FError {
    #[error("{0}")]
    Generic(String),
    #[error("HTTP request failed: {0}")]
    HttpError(String),
    #[error("Failed to parse response: {0}")]
    ParseError(String),
    #[error("Config Error: {0}")]
    ConfigError(String),
    #[error("Context Error: {0}")]
    ContextError(String),
    #[error("Context Error: {0}")]
    IoError(String),
}

pub trait CallContext: Send + Sync {
    fn get(&self, key: String) -> Result<Box<dyn Any>, FError>;
}
pub type ContextBox = Arc<Box<dyn CallContext>>;

struct EmptyCallContext;
impl CallContext for EmptyCallContext {
    fn get(&self, _: String) -> Result<Box<dyn Any>, FError> {
        Err(FError::ContextError("key not found".to_string()))
    }
}
pub fn make_empty_ctx() -> ContextBox {
    Arc::new(Box::new(EmptyCallContext))
}

struct ContextBoxUtils;
impl ContextBoxUtils {
    pub fn make_from_fn<F>(f: F) -> impl CallContext
    where
        F: Fn(String) -> Result<Box<dyn Any>, FError> + Send + Sync
    {
        struct InnerCallContext<F> {
            f: F,
        }
        impl<F> CallContext for InnerCallContext<F> 
        where 
            F: Fn(String) -> Result<Box<dyn Any>, FError> + Send + Sync
        {
            fn get(&self, key: String) -> Result<Box<dyn Any>, FError> {
                (self.f)(key)
            }
        }

        InnerCallContext { f }
    }
}


#[async_trait]
pub trait CallCore<T: Send, R: Send> : Send + Sync {
    async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError>;
}
#[async_trait]
pub trait FnCore<T: Send + Sync, R: Send + Sync> : Send + Sync {
    fn run(&self, arg: T) -> Result<R, FError>;
}

pub type CallBox<'a, T, R> = Arc<Box<dyn CallCore<T, R> + 'a>>;
pub type FnBox<'a, T, R> = Box<dyn FnCore<T, R> + 'a>;

pub struct ListHandler<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> {
    parallel_num: usize,
    item_handler: CallBox<'a, T, R>,
}
impl<'a, T: Send + Sync +'a, R: Send + Sync +'a> ListHandler<'a, T, R> {
    pub fn new(parallel_num: usize, item_handler: CallBox<'a, T, R>) -> Self {
        Self {
            parallel_num,
            item_handler,
        }
    }
}
#[async_trait]
impl <'a, T: Send + Sync +'a, R: Send + Sync +'a> CallCore<Vec<T>, Vec<R>> for ListHandler<'a, T, R> {
    async fn run(&self, arg: Vec<T>, ctx: ContextBox) -> Result<Vec<R>, FError> {
        use futures::stream::{self, StreamExt};
        
        // 使用 StreamExt::buffer_unordered 来限制并行数量
        let results = stream::iter(arg)
            .map(|item| {
                let handler = &self.item_handler;
                let ctx = ctx.clone();
                async move { handler.run(item, ctx).await }
            })
            .buffered(self.parallel_num) // 使用 parallel_num 控制并行度
            .collect::<Vec<Result<R, FError>>>()
            .await;
        
        // 处理结果
        let mut output = Vec::with_capacity(results.len());
        for result in results {
            match result {
                Ok(r) => output.push(r),
                Err(e) => return Err(e),
            }
        }

        Ok(output)
    }
}

pub struct Composer {
}

impl Composer {
    pub fn from_fn<'a, T: Send + Sync + 'a, R: Send + Sync + 'a, F, Fut>(f: F) -> CallBox<'a, T, R>
    where
        Fut: Future<Output = Result<R, FError>> + Send + 'a,
        F: Fn(T) -> Fut + Send + Sync + 'a,
    {
        Self::from_fn2(move |arg, _ctx| f(arg))
    }

    pub fn from_fn2<'a, T: Send + Sync + 'a, R: Send + Sync + 'a, F, Fut>(f: F) -> CallBox<'a, T, R>
    where
        Fut: Future<Output = Result<R, FError>> + Send + 'a,
        F: Fn(T, ContextBox) -> Fut + Send + Sync + 'a,
    {
        struct FnRunable1<T, R, F> {
            f: F,
            _phantom: std::marker::PhantomData<(T, R)>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R: Send + Sync, F, Fut> CallCore<T, R> for FnRunable1<T, R, F>
        where
            Fut: Future<Output = Result<R, FError>> + Send + 'a,
            F: Fn(T, ContextBox) -> Fut + Send + Sync,
        {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError> {
                (self.f)(arg, ctx).await
            }
        }

        Arc::new(Box::new(FnRunable1 {
            f,
            _phantom: std::marker::PhantomData,
        }))
    }

    pub fn identity<'a, T: Send + Sync +'a>() -> CallBox<'a, T, T> {
        Self::from_fn(|arg: T| Box::pin(async move { Ok(arg) })) 
    }

    pub fn with_context_params<'a, T, R>(key: String, call: CallBox<'a, T, R>) -> CallBox<'a, T, R> 
    where 
        T: Send + Sync + Clone +'static,
        R: Send + Sync +'a
    {        
        let call_arc = Arc::new(call);
        
        Self::from_fn2(move |arg: T, ctx: ContextBox| {
            let arg_clone = arg.clone();
            let key_clone = key.clone();
            let ctx = ContextBoxUtils::make_from_fn(move |key_arg: String| {
                if key_clone == key_arg {
                    Ok(Box::new(arg_clone.clone()))
                } else {
                    ctx.get(key_arg)
                }
            });

            let call = call_arc.clone();
            Box::pin(async move {
                // 解引用 Arc 获取内部的 CallBox
                call.run(arg, Arc::new(Box::new(ctx))).await
            })
        })
    }


    pub fn with_callbox<T: Send + Sync + Clone + 'static>(key: &str) -> CallBox<'static, (), T> {
        let key_str = key.to_string();
        
        struct WithHandler<T: Clone + Send + Sync + 'static> {
            key: String,
            _phantom: std::marker::PhantomData<T>,
        }
        
        #[async_trait]
        impl<T: Clone + Send + Sync + 'static> CallCore<(), T> for WithHandler<T> {
            async fn run(&self, _: (), _ctx: ContextBox) -> Result<T, FError> {
                todo!()
            }
        }
        
        Arc::new(Box::new(WithHandler {
            key: key_str,
            _phantom: std::marker::PhantomData,
        }))
    }


    pub fn compose<'a, T, R, T2> (f1: CallBox<'a, T, T2>, f2: CallBox<'a, T2, R>) -> CallBox<'a, T, R>
    where 
        T: Send + Sync + 'a,
        R: Send + Sync + 'a,
        T2: Send + Sync + 'a,
    {
        struct ComposedRunable<'a, T, R, T2> {
            f1: CallBox<'a, T, T2>,
            f2: CallBox<'a, T2, R>,
        }

        #[async_trait]
        impl<'a, T, R, T2> CallCore<T, R> for ComposedRunable<'a, T, R, T2>
        where 
            T: Send + Sync, 
            R: Send + Sync, 
            T2: Send + Sync 
        {
            async fn run(&self, t: T, ctx: ContextBox) -> Result<R, FError> {
                let intermediate = self.f1.run(t, ctx.clone()).await?;
                self.f2.run(intermediate, ctx).await
            }
        }

        Arc::new(Box::new(ComposedRunable { f1, f2 }))
    }

    pub fn compose3<'a, T, R, T1, T2> (
        f1: CallBox<'a, T, T1>, 
        f2: CallBox<'a, T1, T2>, 
        f3: CallBox<'a, T2, R>
    ) -> CallBox<'a, T, R>
    where 
        T: Send + Sync + 'static, 
        R: Send + Sync + 'static, 
        T1: Send + Sync + 'static, 
        T2: Send + Sync + 'static
    {
        let composed = Self::compose(f1, f2);
        Self::compose(composed, f3)
    }

    pub fn make_exec<T: Send + Sync + 'static, R: Send + Sync + 'static>() -> CallBox<'static, (CallBox<'static, T, R>, T), R> {
        struct FnA {}
        #[async_trait]
        impl<T: Send + Sync + 'static, R: Send + Sync + 'static> CallCore<(CallBox<'static, T, R>, T), R> for FnA  {
            async fn run(&self, (f, arg): (CallBox<'static, T, R>, T), ctx: ContextBox) -> Result<R, FError> {
                f.run(arg, ctx).await
            }
        }
        Arc::new(Box::new(FnA {}))
    }

    pub fn compose_parallel_base<'a, T1, T2, R1, R2>(
        f1: CallBox<'a, T1, R1>,
        f2: CallBox<'a, T2, R2>,
    ) -> CallBox<'a, (T1, T2), (R1, R2)>
    where
        T1: Send + Sync + 'a,
        T2: Send + Sync + 'a,
        R1: Send + Sync + 'a,
        R2: Send + Sync + 'a,
    {
        struct ParallelRunable<'a, T1, T2, R1, R2> {
            f1: CallBox<'a, T1, R1>,
            f2: CallBox<'a, T2, R2>,
        }

        #[async_trait]
        impl<'a, T1: Send + Sync, T2: Send + Sync, R1: Send + Sync, R2: Send + Sync> CallCore<(T1, T2), (R1, R2)> 
            for ParallelRunable<'a, T1, T2, R1, R2> 
        {
            async fn run(&self, t: (T1, T2), ctx: ContextBox) -> Result<(R1, R2), FError> {
                let (t1, t2) = t;
                
                let (r1, r2) = tokio::join!(
                    self.f1.run(t1, ctx.clone()),
                    self.f2.run(t2, ctx)
                );

                match (r1, r2) {
                    (Ok(r1), Ok(r2)) => Ok((r1, r2)),
                    (Err(e), _) => Err(e),
                    (_, Err(e)) => Err(e),
                }
            }
        }

        Arc::new(Box::new(ParallelRunable { f1, f2 }))
    }

    pub fn compose_parallel<'a, T, R1, R2>(
        f1: CallBox<'a, T, R1>,
        f2: CallBox<'a, T, R2>,
    ) -> CallBox<'a, T, (R1, R2)>
    where
        T: Send + Sync + Clone + 'a,
        R1: Send + Sync + 'a,
        R2: Send + Sync + 'a,
    {
        let f_base = Self::compose_parallel_base(f1, f2);
        struct ParallelRunable<'a, T, R1, R2> {
            f_base: CallBox<'a, (T, T), (R1, R2)>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync + Clone, R1: Send + Sync, R2: Send + Sync> CallCore<T, (R1, R2)>
            for ParallelRunable<'a, T, R1, R2>
        {
            async fn run(&self, t: T, ctx: ContextBox) -> Result<(R1, R2), FError> {
                let t_copy = t.clone();
                let (r1, r2) = self.f_base.run((t_copy, t), ctx).await?;
                Ok((r1, r2))
            }
        }

        Arc::new(Box::new(ParallelRunable { f_base }))
    }

    pub fn compose_parallel_simple<'a, T, R1, R2>(
        f1: CallBox<'a, (), R1>,
        f2: CallBox<'a, T, R2>,
    ) -> CallBox<'a, T, (R1, R2)>
    where
        T: Send + Sync  + 'a,
        R1: Send + Sync + 'a,
        R2: Send + Sync + 'a,
    {
        let f_base = Self::compose_parallel_base(f1, f2);
        struct ParallelRunable<'a, T, R1, R2> {
            f_base: CallBox<'a, ((), T), (R1, R2)>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R1: Send + Sync, R2: Send + Sync> CallCore<T, (R1, R2)>
            for ParallelRunable<'a, T, R1, R2>
        {
            async fn run(&self, t: T, ctx: ContextBox) -> Result<(R1, R2), FError> {
                let (r1, r2) = self.f_base.run(((), t), ctx).await?;
                Ok((r1, r2))
            }
        }

        Arc::new(Box::new(ParallelRunable { f_base }))
    }

    
    fn make_recursive<'a, T, R>(fable: CallBox<'a, (CallBox<'a, T, R>, T), R>) -> CallBox<'a, T, R>
    where
        T: Send + Sync + 'a,
        R: Send + Sync + 'a,
    {
        struct FnA<'a, T, R> {
            fable: Arc<CallBox<'a, (CallBox<'a, T, R>, T), R>>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> CallCore<T, R> for FnA<'a, T, R> {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError> {
                // 创建一个自引用的函数对象
                let self_ref: CallBox<T, R> = Arc::new(Box::new(Self {
                    fable: self.fable.clone(),
                }));
                
                // 调用 fable 并传入自身引用和参数
                self.fable.run((self_ref, arg), ctx).await
            }
        }

        Arc::new(Box::new(FnA {
            fable: Arc::new(fable), 
        }))
    }

    pub fn wrap_some<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> (
        f: CallBox<'a, T, R>,
    ) -> CallBox<'a, T, Option<R>> {
        struct SomeRunable<'a, T, R> {
            f: Arc<CallBox<'a, T, R>>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R: Send + Sync> CallCore<T, Option<R>> for SomeRunable<'a, T, R> {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<Option<R>, FError> {
                let rs = self.f.run(arg, ctx).await?;
                Ok(Some(rs))
            }
        }

        Arc::new(Box::new(SomeRunable {
            f: Arc::new(f),
        }))
    }

    pub fn wrap_option<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> (
        fsome: CallBox<'a, T, R>, 
        fnone: CallBox<'a, (), R>
    ) -> CallBox<'a, Option<T>, R> {
        struct OptionRunable<'a, T, R> {
            fsome: CallBox<'a, T, R>, 
            fnone: CallBox<'a, (), R>
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R: Send + Sync> CallCore<Option<T>, R> for OptionRunable<'a, T, R> {
            async fn run(&self, t: Option<T>, ctx: ContextBox) -> Result<R, FError> {
                match t {
                    Some(value) => self.fsome.run(value, ctx).await,
                    None => self.fnone.run((), ctx).await,
                }
            }
        }

        Arc::new(Box::new(OptionRunable { fsome, fnone }))
    }
}






#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_code() {
        fn make_parallel_executor<'a, T: Send + Sync + 'a, R: Send + Sync + 'a>(parallel_num: usize) -> FnBox<'a, CallBox<'a, T, R>, CallBox<'a, T, R>> {
            todo!()
        }

        fn make_vec_handler<'a, T: Send + Sync + 'a + Clone, R: Send + Sync + 'a>(
            parallel_executor: FnBox<'a, CallBox<'a, T, R>, CallBox<'a, T, R>>,
            item_handler: CallBox<'a, T, R>
        ) -> CallBox<'a, Vec<T>, Vec<R>> {
            todo!()
        }


        let parallel_executor = make_parallel_executor(2);
        let item_handler = Composer::from_fn(async |s: String| Ok(format!("{}: handle", s)));

        let vec_handler = make_vec_handler(parallel_executor, item_handler);
        
        let data_callbox = Composer::from_fn(async|()| Ok(vec!["1".to_string(), "2".to_string(), "3".to_string()]));
        let composed = Composer::compose(data_callbox, vec_handler);

        let ctx = make_empty_ctx();
        let result = composed.run((), ctx).await;
        assert_eq!(result, Ok(vec!["1: handle".to_string(), "2: handle".to_string(), "3: handle".to_string()]))
    }

    #[tokio::test]
    async fn test_from_fn() {
        async fn fn_a(arg: String) -> Result<String, FError> {
            Ok(format!("{}: A", arg))
        }
        let composed = Composer::from_fn(fn_a);

        let composed = Composer::compose(
            Composer::compose(
                composed, 
                Composer::from_fn(async |s| Ok(((1, 3), (s, "88"), (3.6), (ListHandler::new(3, Composer::from_fn(fn_a))))))
            ), 
            Composer::from_fn(async |args| {
                let (_, (s, _), _, _) = args;
                Ok(format!("{}", s))
            })
        );


        let result = composed.run("test".to_string(), make_empty_ctx()).await;
        assert_eq!(result, Ok("test: A".to_string()));
    }

    #[tokio::test]
    async fn test_from_fn2() {
        fn make_exec<T: Send + Sync + 'static, R: Send + Sync + 'static>() -> CallBox<'static, (CallBox<'static, T, R>, T), R> {
            struct FnA {}
            #[async_trait]
            impl<T: Send + Sync + 'static, R: Send + Sync + 'static> CallCore<(CallBox<'static, T, R>, T), R> for FnA  {
                async fn run(&self, (f, arg): (CallBox<'static, T, R>, T), ctx: ContextBox) -> Result<R, FError> {
                    f.run(arg, ctx).await
                }
            }
            Arc::new(Box::new(FnA {}))
        }
        
    }

    #[tokio::test]
    async fn test_option() {
        struct FnA {}
        struct FnB {}  
        #[async_trait]
        impl CallCore<String, String> for FnA  {
            async fn run(&self, arg: String, _: ContextBox) -> Result<String, FError> {
                Ok(format!("{}: A", arg))
            }
    
        }

        #[async_trait]
        impl CallCore<(), String> for FnB  {
            async fn run(&self, _: (), _: ContextBox) -> Result<String, FError> {
                Ok("None".to_string())
            }
        }
        let composed = Composer::wrap_option(Arc::new(Box::new(FnA {})), Arc::new(Box::new(FnB {})));
        let result = composed.run(Some("test".to_string()), make_empty_ctx()).await;
        assert_eq!(result, Ok("test: A".to_string()));
    }

    #[tokio::test]
    async fn test_ss() {

        struct AStruct {}
        #[async_trait]
        impl CallCore<(CallBox<'static, i32, i32>, i32), i32> for AStruct {
            async fn run(&self, (f, x): (CallBox<'static, i32, i32>, i32), ctx: ContextBox) -> Result<i32, FError> {
                let rs = match x {
                    0 => 1,
                    _ => x * f.run(x - 1, ctx).await?,
                };
                Ok(rs)
            }
        }
        let factorial = Composer::make_recursive(Arc::new(Box::new(AStruct{})));

        let result = factorial.run(5, make_empty_ctx()).await.unwrap();
        assert_eq!(result, 120);
        // let fib = r(boxed_fib);
    }

    #[test]
    fn test_y_combine() {
        fn r<A, R>(g: impl Fn(&dyn Fn(A) -> R, A) -> R) -> impl Fn(A) -> R {
            fn r_inner<'a, A, R>(g: &'a dyn Fn(&dyn Fn(A) -> R, A) -> R) -> impl Fn(A) -> R + 'a {
                move |x| g(&r_inner(g), x)
            }
            move |x| r_inner(&g)(x)
        }
        let g = |f: &dyn Fn(usize) -> usize, x| match x {
            0 => 1,
            _ => x * f(x - 1),
        };

        let fact = r(g);
        println!("{}", fact(5)); // 将会输出 120
    }

    #[tokio::test]
    async fn test_composer() {
        struct FnA {}
        struct FnB {}

        #[async_trait]
        impl CallCore<String, String> for FnA  {
            async fn run(&self, arg: String, _: ContextBox) -> Result<String, FError> {
                Ok(format!("{}: A", arg)) 
            }
        }

        #[async_trait]
        impl CallCore<String, String> for FnB  {
            async fn run(&self, arg: String, _: ContextBox) -> Result<String, FError> {
                Ok(format!("{}: B", arg))
            } 
        }
        let composed = Composer::compose(Arc::new(Box::new(FnA {})), Arc::new(Box::new(FnB {})));
        let result = composed.run("test".to_string(), make_empty_ctx()).await;
        assert_eq!(result, Ok("test: A: B".to_string()));
    }
}
